package com.gitee.huxing.courese.demo;

import com.gitee.huxing.courese.core.KmqBroker;
import com.gitee.huxing.courese.core.KmqConsumer;
import com.gitee.huxing.courese.core.KmqMessage;
import com.gitee.huxing.courese.core.KmqProducer;
import lombok.SneakyThrows;

/**
 * @Description:
 * @ProjectName: JavaCourse
 * @Package: com.gitee.huxing.courese.demo
 * @ClassName: KmqDemo
 * @Author: huxing
 * @DateTime: 2021-11-07 4:12 下午
 */
public class KmqDemo {

    private static final int CAPACITY = 1000;

    @SneakyThrows
    public static void main(String[] args) {
        String topic = "kk.test";
        KmqBroker broker = new KmqBroker();
        broker.createTopic(topic);

        // TODO: 创建消费者处理线程
        KmqConsumer consumer = broker.createConsumer();
        consumer.subscribe(topic);
        final boolean[] flag = new boolean[1];
        flag[0] = true;
        new Thread(() ->{
           while (flag[0]){
               KmqMessage<Order> message = consumer.poll(100);
               if (null != message){
                   System.out.println(message.getBody());
               }
           }
           System.out.println("程序退出。");
        }).start();

        // TODO: 创建生产者消息创建线程
        KmqProducer producer = broker.createProducer();
        for (int i = 0; i < CAPACITY; i++) {
            Order order = new Order(1000L + i, System.currentTimeMillis(),
                    "USD2CNY", 6.51d);
            producer.send(topic, new KmqMessage(null, order));
        }
        Thread.sleep(500);
        System.out.println("点击任何键，发送一条消息；点击q或e，退出程序。");
        while (true) {
            char c = (char) System.in.read();
            if(c > 20) {
                System.out.println(c);
                producer.send(topic, new KmqMessage(null, new Order(100000L + c,
                        System.currentTimeMillis(), "USD2CNY", 6.52d)));
            }

            if( c == 'q' || c == 'e') break;
        }

        flag[0] = false;
    }
}
